package com.iot.app.flink.sink;

import com.datastax.driver.mapping.Mapper;
import com.iot.app.flink.vo.CassandraProp;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.cassandra.CassandraSink;

public class RichSink {

    private static CassandraProp cassandraProp;

    public static void initSink(CassandraProp prop) {

        if (prop == null) {
            throw new NullPointerException("Cassandra Prop is null");
        }
        cassandraProp = prop;
    }

    public static<T> void outPutToCassandra(DataStream<T> dataStream) throws Exception {
        CassandraSink.addSink(dataStream)
                .setHost(cassandraProp.getHost())
                .setMapperOptions(() -> new Mapper.Option[]{Mapper.Option.saveNullFields(true)})
                .build();
    }

}
